package com.atguigu.realtime.app.dwd.db;

import com.atguigu.realtime.utils.HbaseUtil;
import com.atguigu.realtime.utils.MyKafkaUtil;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

/**
 * @author: 洛尘
 * @since: 2023-10-08 08:06
 * @description: 评论事实表
 **/
public class DwdInteractionCommentInfo {
    public static void main(String[] args) {
        //1.基本环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        //1.2设置并行度
        env.setParallelism(4);
        //1.3创建表执行环境
        StreamTableEnvironment tableEnv = StreamTableEnvironment.create(env);
        //2.检查点略
        //3.从kafka主题中读取数据 创建动态表  --kafka连接器
        //3.1声明消费者组
        String grouId="dwd_interaction_comment_group";
        //3.2执行建表
        tableEnv.executeSql(MyKafkaUtil.getTopicDbDDL(grouId));
        //4.将评论数据过滤处来 --tablename --type --table='comment_info' and type='insert'
        Table commentTable = tableEnv.sqlQuery("select \n" +
                "    `data`['id'] id,\n" +
                "    `data`['user_id'] user_id,\n" +
                "    `data`['sku_id'] sku_id,\n" +
                "    `data`['appraise'] appraise,\n" +
                "    `data`['comment_txt'] comment_txt,\n" +
                "    ts,\n" +
                "    proc_time\n" +
                "from topic_db where `table`='comment_info' and `type`='insert'");

        tableEnv.createTemporaryView("comment_table",commentTable);
//        tableEnv.executeSql("select * from comment_table").print();
        //5.从hbase 中读取字典表数据 创建动态表    hbase连接器
        tableEnv.executeSql(HbaseUtil.getBaseDicDDL());
        //6.关联评论表和字典表   用的是lookupjoin
        Table joinedTable = tableEnv.sqlQuery("SELECT\n" +
                " id,\n" +
                " user_id,\n" +
                " sku_id,\n" +
                " appraise,\n" +
                " dic.dic_name appraise_name,\n" +
                " comment_txt,\n" +
                " ts\n" +
                "FROM comment_table AS c\n" +
                "  JOIN base_dic FOR SYSTEM_TIME AS OF c.proc_time AS dic\n" +
                "    ON c.appraise = dic.dic_code");
        tableEnv.createTemporaryView("joined_table",joinedTable);
//        tableEnv.executeSql("select * from joined_table").print();
        //7.将关联的结果写入到kafka主题   --连接器的选择是upsert kafka
        // 7.1创建动态表和要写入的主题进行关联
        tableEnv.executeSql("CREATE TABLE dwd_interaction_comment (\n" +
                "     id string,\n" +
                "     user_id string,\n" +
                "     sku_id string,\n" +
                "     appraise string,\n" +
                "     appraise_name string,\n" +
                "     comment_txt string,\n" +
                "     ts string,\n" +
                "  PRIMARY KEY (id) NOT ENFORCED\n" +
                ") " + MyKafkaUtil.getUpsertKafkaDDL("dwd_interaction_comment"));
        //7.2 写入
//        tableEnv.executeSql("select * from joined_table").print();
        tableEnv.executeSql("insert into dwd_interaction_comment select * from joined_table");

    }
}